package com.csw.flink.transformation

import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo02FlatMap {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    val ds: DataStream[String] = env.socketTextStream("master", 8888)

    //scala的方式，函数返回一个数组
    val ds1: DataStream[String] = ds.flatMap(value => value.split(","))

    //java的方式  使用匿名内部类
    var ds2: DataStream[String] = ds.flatMap(new FlatMapFunction[String, String] {
      override def flatMap(value: String, out: Collector[String]): Unit = {

        val split: Array[String] = value.split(",")

        for (elem <- split) {
          //将数据发送到下游
          out.collect(elem)
        }
      }
    })

    //scala 升级版
    var ds3: DataStream[String] = ds.flatMap((value: String, out: Collector[String]) => {
      val split: Array[String] = value.split(",")

      for (elem <- split) {

        //将数据发送到下游
        out.collect(elem)
      }
    })

    ds3.print()

    env.execute()
  }
}
